package com.irt.rootcloud.realtime.gketl.bolt;

import java.text.DecimalFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.commons.lang.StringUtils;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.irt.rootcloud.realtime.gketl.model.LocationInfo;
import com.irt.rootcloud.realtime.gketl.util.HttpRequest;
import com.irt.rootcloud.realtime.gketl.util.LogParseTool;
import com.irt.rootcloud.realtime.gketl.util.Utils;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoClientURI;
import com.mongodb.MongoException;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

/**
 * <p>
 * 将数据处理处理并翻译经纬度
 * </p>
 * 
 * @author Ken.zhang
 * @date 2017-1-18 下午2:28:05
 */
public class EtlBolt extends BaseBasicBolt {
	private static final long serialVersionUID = -5424283134617576073L;
	private Properties prop;
	private String latname;
	private String lontname;
	private LocationInfo locationInfo;
	private String tennatStr;
	private String deviceTypeStr;
	private String deviceId;
	private static DecimalFormat df = new DecimalFormat("######0.0000");
	private static SimpleDateFormat sf1 = new SimpleDateFormat(
			"yyyy-MM-dd HH:mm:ss.SSS");

	public DB myMongo;
	
	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		super.cleanup();
	}

	public EtlBolt(Properties prop) {
		this.prop = prop;
	}

	private String url;

	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		// TODO Auto-generated method stub
		super.prepare(stormConf, context);
		latname = prop.getProperty("schema.lat.name").toString();
		lontname = prop.getProperty("schema.lont.name").toString();
		url = prop.getProperty("gps.location.url").toString();
		
		initDBPrompties();
	}

	public void execute(Tuple tuple, BasicOutputCollector collector) {
		// TODO Auto-generated method stub
		String dtjson = (String) tuple.getValue(0);
		JSONObject object = JSONObject.parseObject(dtjson);
		tennatStr = object.getString("tenantId");
		deviceTypeStr = object.getString("deviceTypeId");
		deviceId = object.getString("deviceId");
		JSONObject jsonobject = object.getJSONObject("status");
		jsonobject.put("deviceId", deviceId);
		double gsta = 0;
		double gslon = 0;
		try {
			if(jsonobject.getString(latname)==null)
			{
				gsta = 0;
			}else{
			gsta = Double.valueOf(df.format(Double.valueOf(jsonobject
					.getString(latname))));
			}
			if(jsonobject.getString(lontname)==null){
				gslon = 0;
			}else{
				gslon = Double.valueOf(df.format(Double.valueOf(jsonobject
						.getString(lontname))));
			}
		} catch (Exception e) {
			System.out.println(jsonobject.getString(latname) + ","
					+ jsonobject.getString(lontname));
//			System.out.println(dtjson);
			e.printStackTrace();
		}

//		Map<String, String> params = new HashMap<String, String>();
//		params.put("latitude", String.valueOf(gsta));
//		params.put("longitude", String.valueOf(gslon));
//		String strjson = HttpClientUtil.post(url, params);

		//使用ES获取经纬度。
//		String strjson = getGpsJson(String.valueOf(gsta),String.valueOf(gslon));
//		if (strjson != null && !strjson.trim().equals("")) {
//			locationInfo = getLocationInfo(strjson);
//		} else {
//			locationInfo = new LocationInfo();
//		}

		//使用mongo获取经纬度
		LocationInfo locationInfo = getLocationInfo(gsta,gslon);
		if(locationInfo==null){
			locationInfo= new LocationInfo();
		}
		
		jsonobject.put("provinceId", locationInfo.getProvinceId());
		jsonobject.put("cityId", locationInfo.getCityId());
		jsonobject.put("areaId", locationInfo.getAreaId());
		jsonobject.put("provinceName", locationInfo.getProvinceName());
		jsonobject.put("cityName", locationInfo.getCityName());
		jsonobject.put("areaName", locationInfo.getAreaName());
		try {
			Object localtime = jsonobject.get("Timestamp_Local");
			Object cloudtime = jsonobject.get("Timestamp_CloudM2M");
			if(localtime!=null)
			jsonobject.put("localtime", sf1.parse(localtime.toString()));
			if(cloudtime!=null)
			jsonobject.put("cloudtime", sf1.parse(cloudtime.toString()));

		} catch (ParseException e) {
			System.out.println(jsonobject.getString("Timestamp_Local") + ","
					+ jsonobject.getString("Timestamp_CloudM2M"));
//			System.out.println(dtjson);
			e.printStackTrace();
		}

		Values values = new Values(tennatStr,deviceTypeStr,jsonobject.toString());
		collector.emit(values);
	}

	public void declareOutputFields(OutputFieldsDeclarer declare) {
		// TODO Auto-generated method stub
		declare.declare(new Fields(Utils.CHANNEL_TENANT,Utils.CHANNEL_DEVICETYPE,Utils.FIELDCHANNEL_BOLT));
	}

	private String getGpsJson(String lat,String lont){
//		String url = "http://10.70.17.205:9200/gps2/dis_map/_search";
		String jsonStr = "{\"query\": {\"bool\": {\"must\": [{\"term\":{\"latitude\":"+lat+"}},{\"range\":{\"longitudeStart\":{\"lte\":"+lont+"}}},{\"range\":{\"longitudeEnd\":{\"gte\":"+lont+"}}}]}}}";
		JSONObject jsonParam=JSONObject.parseObject(jsonStr);
		String timeAggsResult = HttpRequest.sendJsonPostUTF8(url, jsonParam.toJSONString());
		
		JSONObject result = JSONObject.parseObject(timeAggsResult);
		
		JSONArray ja = result.getJSONObject("hits").getJSONArray("hits");
		if(ja == null || ja.size() == 0){
			return null;
		}else{
			return ja.getJSONObject(0).get("_source").toString();
		}
		
//	    String gpsjson = JSONObject.parseObject(JSONArray.parseArray(JSONObject.parseObject(result.get("hits").toString()).get("hits").toString()).get(0).toString()).get("_source").toString();
//	    return gpsjson;
	}
	
	private LocationInfo getLocationInfo(String str) {
		LocationInfo locinfo = new LocationInfo();
		try {
			Map<String, String> map = LogParseTool.jsonToMap(str);
			locinfo = (LocationInfo) LogParseTool.convLogToBean(map,
					LocationInfo.class);

		} catch (Exception e) {
			e.printStackTrace();
		}
		return locinfo;
	}
	
	
	// 根据经纬度查询省市区
	private LocationInfo getLocationInfo(Double lat, Double lont) {
		LocationInfo locinfo = new LocationInfo();
		try{
		
		DBObject dbObject = new BasicDBObject();
		// 经度
		dbObject.put("longitudeStart", new BasicDBObject("$lte", lont));
		dbObject.put("longitudeEnd", new BasicDBObject("$gte", lont));
		// 维度
		dbObject.put("latitude", lat);
		List<DBObject> list = find(prop.get("mongo.db.localtion.tablename")
				.toString(), dbObject);
		if (list.size() > 0) {
			Map<String, String> map = LogParseTool.jsonToMap(list.get(0)
					.toString());
			locinfo = (LocationInfo) LogParseTool.convLogToBean(
					map, LocationInfo.class);
		}
		}catch(Exception e){
			e.printStackTrace();
		}
		return locinfo;
	}
	
	public List<DBObject> find(String collection, DBObject obj) {
		DBCollection coll = getDBCollection(collection);
		DBCursor c = coll.find(obj);
		if (c != null)
			return c.toArray();
		else
			return null;
	}

	public DBCollection getDBCollection(String tableName) {
		DBCollection collection = myMongo.getCollection(tableName);
		return collection;
	}
	
	/**
	 * 初始化连接池 ./bin/mongo 192.168.1.15:27017/uone -uuone2 -puone2
	 */
	@SuppressWarnings("deprecation")
	private void initDBPrompties() {
		// 其他参数根据实际情况进行添加
		try {
			MongoClientOptions.Builder build = new MongoClientOptions.Builder();
			build.connectionsPerHost(5); // 与目标数据库能够建立的最大connection数量为50
			// build.autoConnectRetry(true); // 自动重连数据库启动
			build.threadsAllowedToBlockForConnectionMultiplier(2); // 如果当前所有的connection都在使用中，则每个connection上可以有50个线程排队等待
			/*
			 * 一个线程访问数据库的时候，在成功获取到一个可用数据库连接之前的最长等待时间为2分钟
			 * 这里比较危险，如果超过maxWaitTime都没有获取到这个连接的话，该线程就会抛出Exception
			 * 故这里设置的maxWaitTime应该足够大，以免由于排队线程过多造成的数据库访问失败
			 */
			build.maxWaitTime(1000 * 60 * 2);
			build.connectTimeout(1000 * 60 * 1); //
			MongoClientURI uri = new MongoClientURI(prop.get("mongo.url")
					.toString(), build);
			// MongoClientURI uri = new MongoClientURI(
			// "mongodb://uone:uone#456Fyd@10.47.210.127:27017/uone", build);
			MongoClient mongo = new MongoClient(uri);
			myMongo = mongo.getDB(prop.get("mongo.db.name").toString());
		} catch (MongoException e) {
			e.printStackTrace();
		} catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
		}

	}
	
	public static void main(String args[]){
		String aa = "";
		boolean b = StringUtils.isBlank(aa);
		System.out.println(b);
	}
}
